package com.witsight.cassandra.dao;

import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicReference;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.cassandra.core.CassandraOperations;
import org.springframework.data.cassandra.mapping.CassandraMappingContext;
import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
import org.springframework.data.cassandra.mapping.CassandraPersistentProperty;
import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
import org.springframework.data.cassandra.repository.query.CassandraEntityMetadata;
import org.springframework.data.cassandra.repository.query.CassandraParameterAccessor;
import org.springframework.data.cassandra.repository.query.CassandraQueryMethod;
import org.springframework.data.cassandra.repository.query.PartTreeCassandraQuery;
import org.springframework.data.cassandra.repository.query.StringBasedCassandraQuery;
import org.springframework.data.cassandra.repository.support.CassandraRepositoryFactory;
import org.springframework.data.cassandra.repository.support.SimpleCassandraRepository;
import org.springframework.data.mapping.PropertyHandler;
import org.springframework.data.projection.ProjectionFactory;
import org.springframework.data.repository.core.NamedQueries;
import org.springframework.data.repository.core.RepositoryInformation;
import org.springframework.data.repository.core.RepositoryMetadata;
import org.springframework.data.repository.query.EvaluationContextProvider;
import org.springframework.data.repository.query.QueryLookupStrategy;
import org.springframework.data.repository.query.QueryLookupStrategy.Key;
import org.springframework.data.repository.query.RepositoryQuery;
import org.springframework.data.repository.query.parser.Part;
import org.springframework.data.repository.query.parser.PartTree;
import org.springframework.expression.spel.standard.SpelExpressionParser;

import com.witsight.cassandra.dao.impl.DynamicCassandraRepository;
import com.witsight.platform.dao.CommonRepository;
import com.witsight.platform.util.lang.StringUtil;

/**   
 * 说明：自定义Cassandra仓库工厂类
 * @Title: CustomCassandraRepositoryFactory.java 
 * @Package com.witsight.cassandra.dao 
 * @See: {@link CassandraRepositoryFactory}
 * Copyright: Copyright (c) 2017
 * Company:sany huax witsight team by product
 * @author: penght  
 * @date: 2017年9月26日 上午11:35:22 
 * @version: V1.0
 *
 */
public class CustomCassandraRepositoryFactory extends CassandraRepositoryFactory {
	/**
     * Logger
     */
    private static final Logger log = LoggerFactory.getLogger(CustomCassandraRepositoryFactory.class);
    private static final SpelExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
    
    private final CassandraOperations operations;    
	private final CassandraMappingContext mappingContext;
	private NameQueriesRepository nameQueries;
	private final CommonRepository commonRepository ;
	/**
	 * 
	 * 说明：构造函数 
	 * @Title:CustomCassandraRepositoryFactory.java
	 * @See: {@link #CustomCassandraRepositoryFactory(CassandraOperations)}
	 * @param operations
	 * @date 2017年9月26日 上午11:35:41
	 */
	public CustomCassandraRepositoryFactory(CassandraOperations operations,CommonRepository commonRepository) {
		super(operations);
		this.operations = operations;
		this.mappingContext = operations.getConverter().getMappingContext();
		this.commonRepository = commonRepository;
	}
	
	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getRepositoryBaseClass(org.springframework.data.repository.core.RepositoryMetadata)
	 */
	@Override
	protected Class<?> getRepositoryBaseClass(RepositoryMetadata metadata) {
		if (isCustomDynamicQuery(metadata.getRepositoryInterface())) {
			return BaseCassandraRepository.class;
		}
		return SimpleCassandraRepository.class;
	}
	
	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	protected Object getTargetRepository(RepositoryInformation information) {
		Class<?> repositoryInterface = information.getRepositoryInterface();
		CassandraEntityInformation<?, Serializable> entityInformation = getEntityInformation(information.getDomainType());
		if (isCustomDynamicQuery(repositoryInterface)) {
			if (log.isDebugEnabled()) {
				log.info("{}'s class be instanced DynamicCassandraRepository." , repositoryInterface);
			}
			return new DynamicCassandraRepository(entityInformation, this.operations, this.nameQueries, this.commonRepository);
		}
		return super.getTargetRepository(information);
	}
	
	@Override
	public void setNamedQueries(NamedQueries namedQueries) {
		if (namedQueries instanceof NameQueriesRepository) {
			this.nameQueries = (NameQueriesRepository) namedQueries;
		}
		super.setNamedQueries(namedQueries);
	}
	/**
	 * 
	 * 说明：判断是否需要动态sql语句
	 * @Title: isCustomDynamicQuery
	 * @See: 判断是否需要动态sql语句
	 * @author penght 
	 * @param repositoryInterface 接口
	 * @return  true ： 正确  false：不正确
	 * @date: 2015年4月22日 下午12:21:56
	 */
	private boolean isCustomDynamicQuery(Class<?> repositoryInterface) {
		return BaseCassandraRepository.class.isAssignableFrom(repositoryInterface);
	}
	
	/*
	 * (non-Javadoc)
	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getQueryLookupStrategy(Key, EvaluationContextProvider)
	 */
	@Override
	protected QueryLookupStrategy getQueryLookupStrategy(Key key, EvaluationContextProvider evaluationContextProvider) {
		return new CassandraQueryLookupStrategy(operations, evaluationContextProvider, mappingContext);
	}

	private class CassandraQueryLookupStrategy implements QueryLookupStrategy {

		private final EvaluationContextProvider evaluationContextProvider;
		private final CassandraMappingContext mappingContext;
		private final CassandraOperations operations;

		public CassandraQueryLookupStrategy(CassandraOperations operations,
				EvaluationContextProvider evaluationContextProvider, CassandraMappingContext mappingContext) {

			this.operations = operations;
			this.evaluationContextProvider = evaluationContextProvider;
			this.mappingContext = mappingContext;
		}

		/*
		 * (non-Javadoc)
		 * @see org.springframework.data.repository.query.QueryLookupStrategy#resolveQuery(java.lang.reflect.Method, org.springframework.data.repository.core.RepositoryMetadata, org.springframework.data.projection.ProjectionFactory, org.springframework.data.repository.core.NamedQueries)
		 */
		@Override
		public RepositoryQuery resolveQuery(Method method, RepositoryMetadata metadata, ProjectionFactory factory,
				NamedQueries namedQueries) {

			CassandraQueryMethod queryMethod = new CassandraQueryMethod(method, metadata, factory, mappingContext);
			String namedQueryName = queryMethod.getNamedQueryName();

			if (namedQueries.hasQuery(namedQueryName)) {
				String namedQuery = namedQueries.getQuery(namedQueryName);
				return new StringBasedCassandraQuery(namedQuery, queryMethod, operations, EXPRESSION_PARSER,
						evaluationContextProvider);
			} else if (queryMethod.hasAnnotatedQuery()) {
				return new StringBasedCassandraQuery(queryMethod, operations, EXPRESSION_PARSER,
						evaluationContextProvider);
			} else {
				return new CustomPartTreeCassandraQuery(queryMethod, operations);
			}
		}
	}
	
	class CustomPartTreeCassandraQuery extends PartTreeCassandraQuery {
		private final CassandraEntityMetadata<?> cassandraEntityMetadata;
		public CustomPartTreeCassandraQuery(CassandraQueryMethod queryMethod, CassandraOperations operations) {
			super(queryMethod, operations);
			this.cassandraEntityMetadata = queryMethod.getEntityInformation();
		}
		/**
		 * 
		 * 说明：添加允许过滤的标识，判断如果不是主键就需要添加“ALLOW FILTERING”
		 * @Title: addAllFiltering
		 * @See: {@link #addAllFiltering(String, CassandraParameterAccessor)}
		 * @author penght
		 * @param query 
		 * @param parameterAccessor
		 * @return 
		 * @date: 2017年9月27日 下午5:25:34
		 */
		private String addAllFiltering(String query,CassandraParameterAccessor parameterAccessor) {
			PartTree tree = getTree();
			if (tree == null) {
				return query;
			}
			for (Part part : tree.getParts()) {
				final String name = part.getProperty().getSegment();
				if (StringUtil.isEmpty(name)) {
					continue;
				}
				CassandraPersistentEntity<?> entity = mappingContext.getPersistentEntity(cassandraEntityMetadata. getJavaType());
				final AtomicReference<Boolean> hasPrimaryKeyOnlyNotProperties = new AtomicReference<Boolean>(true);
				// 判断是否检索的不是主键
				entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
					@Override
					public void doWithPersistentProperty(CassandraPersistentProperty property) {
						if (!StringUtil.equalsIgnoreCase(property.getName(), name)) {
							return;
						}
						if (property.isCompositePrimaryKey() || property.isPrimaryKeyColumn() || property.isIdProperty()) {
							return;
						}
						hasPrimaryKeyOnlyNotProperties.set(true);
					}
				});
				if (hasPrimaryKeyOnlyNotProperties.get()) {
					query = StringUtil.replace(query, ";", " ALLOW FILTERING;");
					break;
				}
			}
			return query;
		}
		
		@Override
		protected String createQuery(CassandraParameterAccessor parameterAccessor) {
			String query = super.createQuery(parameterAccessor);
			query = addAllFiltering(query , parameterAccessor);
			return query;
		}
	}
}
